package com.gxd.esimport;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class outputThread implements Runnable {

    public BulkProcessor bulkProcessor;
    private TransportClient esclient;
    private LinkedBlockingQueue<Map<String, Object>> queue;

    public outputThread(LinkedBlockingQueue<Map<String, Object>> queue) {
        this.queue = queue;
    }


    public void initESClient() throws NumberFormatException, UnknownHostException {

        Settings settings = Settings.builder()
                .put("cluster.name", "dataservercenteres").build();
        this.esclient = new PreBuiltTransportClient(settings)
                             .addTransportAddress(new TransportAddress(InetAddress.getByName("192.168.13.3"), 9300))
                             .addTransportAddress(new TransportAddress(InetAddress.getByName("192.168.13.4"), 9300));
/*         Setti    ngs settings = Settings.builder()
                .put("cluster.name", "gxdescluster").build();
       this.esclient = new PreBuiltTransportClient(settings)
                             .addTransportAddress(new TransportAddress(InetAddress.getByName("192.168.50.33"), 9300))
                             .addTransportAddress(new TransportAddress(InetAddress.getByName("192.168.50.34"), 9300));*/
        this.bulkProcessor = BulkProcessor.builder(
                 esclient,
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId,
                                           BulkRequest request) {
                    }

                    @Override
                    public void afterBulk(long executionId,
                                          BulkRequest request,
                                          BulkResponse response) {

                        if(response.hasFailures()){
                            throw new RuntimeException("写入elasticsearch出现错误："+response.buildFailureMessage());
                        };

                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void afterBulk(long executionId,
                                          BulkRequest request,
                                          Throwable failure) {
                    }
                })
                .setBulkActions(20000)
                .setBulkSize(new ByteSizeValue(50, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(5)
                .setBackoffPolicy(
                        BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                .build();
    }




    @Override
    public void run() {
        try {

            while(true) {
                    Map event = this.queue.poll(600, TimeUnit.SECONDS);
                    if (event != null){
                        System.out.println(Thread.currentThread().getName()+"出PIPE数据："+String.valueOf(((Map<String, Object>)event.get("rowMap")).get("CITY_CODE"))+"++++"+(String)event.get("index")+"++++"+(String)event.get("index")+"++++"+String.valueOf(((Map<String, Object>)event.get("rowMap")).get("ID")));
                        IndexRequest indexRequest = new IndexRequest((String)event.get("index"),(String)event.get("index"),String.valueOf(((Map<String, Object>)event.get("rowMap")).get("ID"))).source((Map<String, Object>)event.get("rowMap"));
                        this.bulkProcessor.add(indexRequest);
                    }else{
                        System.out.println("写入线程结束："+Thread.currentThread().getName());
                        System.out.println("同步开始时间："+new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date()));
                        System.exit(0);
                    }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

    }


}
